package edu.cmu.arch.pipesandfilters;

import java.io.IOException;
import edu.cmu.arch.pipesandfilters.frame.Frame;
import edu.cmu.arch.pipesandfilters.frame.Variable;

/**
 *  Filters out of range data out of the data stream.
 * 
 *  this filter has two output ports
 *  	OutputPort: outputs the filtered data
 *      OutputPort2: outputs out of range data
 *  
 *  @author Steven Thomas (steventh@andrew.cmu.edu)
 */
public class RangeFilter extends DualOutputFilterFramework {

	
	Variable _Measurement1;
	Variable _Measurement2;    //
	Double _UpperBound1;       // Measurement1's upper bound
	Double _UpperBound2;       // Measurement2's upper bound
	
	/**
	 * Constructor
	 * 
	 * @param Measurement1 
	 * @param UpperBound1   Measurement1's upper bound
	 * @param Measurement2
	 * @param UpperBound2   Measurement2's upper bound
	 */
	public RangeFilter(Variable Measurement1,Double UpperBound1, Variable Measurement2,Double UpperBound2) 
	{
		_Measurement1 = Measurement1;
		_Measurement2 = Measurement2;
		_UpperBound1 = UpperBound1;
		_UpperBound2 = UpperBound2;
    }
	
	
	@Override
	protected void runFilter1() {
		Frame frame = null;
		while (true)
		{
			try
			{
				frame = readFrame();
				processFrame(frame);
		    }
			catch (EndOfStreamException e)
			{
			    //file processing is complete	
				runFilter1Finished=true;
				System.out.print( "\n" + this.getName() + "::Sink Exiting -- \n" );
				break;
	
			} // catch
			catch (IOException ioe)
			{
				System.out.println(this.getName() + "::runFilter1() IOException occured; Unable to read frame" + ioe.getMessage());
				System.out.println(" unable to read frame");
				break;
			}
		}
	}
	
	@Override
	protected void runFilter2() {
		    // Filter2 waits around for filter1 to be done before it closes
		    while(!runFilter1Finished)
		    {
		       try {
					sleep(1);
				} catch (InterruptedException e) {
					System.out.println(this.getName() + "::runFilter2() InterruptedException occured;" + e.getMessage());
					e.printStackTrace();
				}
			}
		    
			ClosePorts();
		    System.out.println("Filter Out Port2 exiting");
		    this.runFilter2Finished = true;
	}

   /**
    * Writes out of range frames to port2
    * @param frame
    * @throws IOException 
    */
   private void writeOutOfRangeFrames(Frame frame) 
   {
	   FrameFilterFramework filter2 = (FrameFilterFramework)this.getOutputFilter2();
	   try
	   {
		   filter2.writeFrame(frame);
	   }
	   catch(IOException e) {
		   System.out.println(this.getName() + "::writeOutOfRangeFrames() IOException occured; Error writing to OutPort 2" + e.getMessage());
	}
   }

    /**
     * Processes a single frame.
     * 	1) If Value is in Range 
	 *      a) Write to port 1
	 *  2) If Value is out of range
     *  	a) Write to port 2
     * @param frame
     */
	private void processFrame(Frame frame) 
	{
		if(isInRange(frame))
		{
			//write to port1
			try {
				writeFrame(frame);
			}catch (IOException e) {
				
				System.err.println("Error writing frame");
				e.printStackTrace();
			}
		}else
		{
			//write to port2
			writeOutOfRangeFrames(frame);
		}
	}
	
	/**
	 * Returns True if data is in the valid range
	 * Returns False if data is outside the valid range
	 * 
	 * @param ValueToCheck a value to check
	 * @return Is value in range
	 */
	private Boolean isInRange(Frame frame)
	{
		Double Measurement1Value = 
			(Double)frame.getValue(_Measurement1);
		Double Measurement2Value =
			(Double)frame.getValue(_Measurement2);
		
		// out of range if both >= _UpperBound 
        if (Measurement1Value <= _UpperBound1 )
        	return true;
        if (Measurement2Value <= _UpperBound2 )
        	return true;
        
        //neither are true, so its not in range.
        return false;
		
	}
}

